package com.sinozo.data.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.sinozo.data.app.function.TradeOrderTimestampsAndWatermarks;
import com.sinozo.data.bean.DwdTradeOrder;
import com.sinozo.data.bean.OrderFiledBean;
import com.sinozo.data.bean.TradeOrderBean;
import com.sinozo.data.common.ConfigConstant;
import com.sinozo.data.common.DataConfig;
import com.sinozo.data.utils.*;
import org.apache.commons.lang.time.DateUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author zhaoyb
 * @version 1.0
 * @description 订单
 * @date 2023/6/14 17:59:02
 */
public class DwsTradeOrderWindowV2 {

    public static void main(String[] args) {

        Logger logger = LoggerFactory.getLogger(DwsTradeOrderWindowV2.class);
        try {
            // 1.获取执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(DataConfig.KAFKA_PARTITION_NUM);
            //1.1 开启CheckPoint
            env.enableCheckpointing(DataConfig.FLINK_CHECKPOINT_INTERVAL, CheckpointingMode.EXACTLY_ONCE);
            env.getCheckpointConfig().setCheckpointTimeout(DataConfig.FLINK_CHECKPOINT_TIMEOUT);
            env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);

            //1.2 设置状态后端
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
            env.getCheckpointConfig().setExternalizedCheckpointCleanup(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
            );
            env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.of(3L, TimeUnit.DAYS), Time.of(1L, TimeUnit.MINUTES)));

            String hdfsPath = DataConfig.HDFS_SERVER;
            env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage(PathUtil.getHdfsPath(hdfsPath, "dwsTradeOrderWindow")));
            env.setStateBackend(new EmbeddedRocksDBStateBackend());

            // 2.读取 Kafka DWD层 订单事实表
            String topicDb = ConfigConstant.KAFKA_DB_TOPIC;
            String topicUserRegister = ConfigConstant.KAFKA_START_LOG_TOPIC;
            String groupId = ConfigConstant.KAFKA_PAY_ORDER_GROUP_ID;

            DataStreamSource<String> topicDbDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topicDb, groupId));
            DataStreamSource<String> topicUserRegisterDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topicUserRegister, groupId));

            // 3.将数据转换为Java对象
            SingleOutputStreamOperator<DwdTradeOrder> topicDbWithDS = topicDbDS.filter(v -> StringUtils.isNotBlank(v)).map(line -> {
                JSONObject jsonObject = JSON.parseObject(line);
                JSONObject data = jsonObject.getJSONObject("data");
                JSONObject old = jsonObject.getJSONObject("old");

                DwdTradeOrder dwdTradeOrder = DwdTradeOrder.builder().id(data.getString("id"))
                        .orderId(data.getString("order_no")).userId(data.getString("user_id"))
                        .deviceId(data.getString("device_id")).productId(data.getString("product_id"))
                        .productName(data.getString("product_name")).orderPrice(AmountUtil.convertFen2Yuan(data.getBigDecimal("amount")))
                        .refundAmount(AmountUtil.convertFen2Yuan(data.getBigDecimal("refund_amount"))).os("")
                        .code("").registerTime(null).table(jsonObject.getString("table")).ts(System.currentTimeMillis())
                        .wayCode(data.getString("way_code")).state(data.getInteger("state")).database(jsonObject.getString("database"))
                        .oldState(old != null ? old.getInteger("state") : null).type(jsonObject.getString("type")).build();
                return dwdTradeOrder;
            });
            topicDbWithDS.print("dbSource>>>>>>>>>>>>");
            OutputTag<String> filterTag = new OutputTag<String>("DbFilter") {
            };
            //过滤数据
            SingleOutputStreamOperator<DwdTradeOrder> filterTopicDbWithDS = topicDbWithDS.process(new ProcessFunction<DwdTradeOrder, DwdTradeOrder>() {
                @Override
                public void processElement(DwdTradeOrder v, Context ctx, Collector<DwdTradeOrder> out) throws Exception {
                    if (validateDb(v)) {
                        out.collect(v);
                    } else {
                        ctx.output(filterTag, JSON.toJSONString(v));
                    }
                }
            });
            //获取侧输出数据并打印
            DataStream<String> dirtyDS = filterTopicDbWithDS.getSideOutput(filterTag);
            dirtyDS.print("DbFilter>>>>>>>>>>>>");

            SingleOutputStreamOperator<DwdTradeOrder> topicUserRegisterWithDS = topicUserRegisterDS.filter(v -> StringUtils.isNotBlank(v)).map(line -> {
                JSONObject jsonObject = JSON.parseObject(line);
                JSONObject common = jsonObject.getJSONObject("common");
                Long ts = jsonObject.getLong("ts");
                Integer startWay = jsonObject.getJSONObject("start").getInteger("start_way");
                String labCode = common.getString("lab_code");
                String labGroupCode = common.getString("lab_group_code");

                DwdTradeOrder bean = DwdTradeOrder.builder().os(common.getString("os")).code(common.getString("code"))
                        .deviceId(common.getString("device_id")).registerTime(new Timestamp(ts)).startWay(startWay).ts(ts)
                        .labCode(labCode).labGroupCode(labGroupCode).build();

                return bean;
            });
            topicUserRegisterWithDS.print("registerUser=>>>>>>>>");

            // 4.将流进行Union
            DataStream<DwdTradeOrder> mergedStream = filterTopicDbWithDS.union(topicUserRegisterWithDS);

            // 4.提取事件时间生成Watermark
            SingleOutputStreamOperator<DwdTradeOrder> jsonObjWithWmDS = mergedStream.assignTimestampsAndWatermarks(new TradeOrderTimestampsAndWatermarks());

            // 5.按照设备分组,使用状态编程提取独立支付用户和首日独立支付用户
            SingleOutputStreamOperator<TradeOrderBean> userPayOrderDS = jsonObjWithWmDS.keyBy(t -> StringUtils.isEmpty(t.getDeviceId()) ? t.getUserId() : t.getDeviceId())
                    .flatMap(new RichFlatMapFunction<DwdTradeOrder, TradeOrderBean>() {
                        //String ===>>> OrderFiledBean
                        private ValueState<OrderFiledBean> lastPayState;

                        @Override
                        public void open(Configuration parameters) throws Exception {
                            StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.days(1))
                                    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                    .build();
                            ValueStateDescriptor<OrderFiledBean> stateDescriptor = new ValueStateDescriptor<>("last-pay-state", OrderFiledBean.class);
                            stateDescriptor.enableTimeToLive(ttlConfig);
                            lastPayState = getRuntimeContext().getState(stateDescriptor);
                        }

                        @Override
                        public void flatMap(DwdTradeOrder tradeOrder, Collector<TradeOrderBean> out) throws Exception {
                            OrderFiledBean orderFiledBean = lastPayState.value();

                            if (StringUtils.isNotBlank(tradeOrder.getProductId())) {

                                //获取状态数据以及当前数据的日期
                                Long dataTime = tradeOrder.getTs();
                                String curDt = DateFormatUtil.toFullDate(dataTime);
                                Timestamp yesToday = new Timestamp(DateUtils.addDays(new Date(), -1).getTime());
                                String firstDt = orderFiledBean != null && StringUtils.isNotBlank(orderFiledBean.getDate()) ? orderFiledBean.getDate() : DateFormatUtil.toDate(yesToday.getTime());
                                //定义当日独立支付人数以及首日付费人数
                                //设备号
                                long deviceNum = 0L;
                                long firstDayDeviceNum = 0L;
                                long firstDayOrderNum = 0L;
                                BigDecimal firstDayPayAmount = BigDecimal.ZERO;
                                boolean firstPay = orderFiledBean != null && orderFiledBean.getExistPay() != null ? orderFiledBean.getExistPay() : false;
                                //判断是否用户已经支付过了
                                if (!firstPay) {
                                    deviceNum = 1L;
                                }
                                //判断是否为当日首次支付
                                if (firstDt.equals(curDt) && tradeOrder.getState().equals(2)) {
                                    //判断是否用户已经支付过了
                                    if (!firstPay) {
                                        firstDayDeviceNum = 1L;
                                    }

                                    firstDayOrderNum = 1l;
                                    firstDayPayAmount = tradeOrder.getOrderPrice();
                                }

                                //金额需要转化为元
                                BigDecimal taxAmount = tradeOrder.getOrderPrice().multiply(TaxUtils.getTax(tradeOrder.getWayCode()));
                                BigDecimal totalAmount = tradeOrder.getOrderPrice().subtract(taxAmount);
                                //业务逻辑处理，当支付成功(state=2)时只计算支付数据、退款成功(state=5)只计算退款数据
                                TradeOrderBean orderBean = null;
                                String createTime = DateFormatUtil.toYmdHms(System.currentTimeMillis());
                                String startTime = DateFormatUtil.toYmdHms(System.currentTimeMillis());
                                String endTime = DateFormatUtil.toYmdHms(DateUtils.addSeconds(new Date(), 10).getTime());

                                if (orderFiledBean == null) {
                                    orderFiledBean = new OrderFiledBean();
                                }
                                if (tradeOrder.getState() == 2) {
                                    //支付数据
                                    orderBean = TradeOrderBean.builder().os(orderFiledBean.getOs()).code(orderFiledBean.getCode()).product_id(tradeOrder.getProductId())
                                            .wayCode(tradeOrder.getWayCode()).payAmount(tradeOrder.getOrderPrice()).refundAmount(BigDecimal.ZERO)
                                            .taxAmount(taxAmount).totalAmount(totalAmount).orderNum(1L).deviceNum(deviceNum)
                                            .firstDayDeviceNum(firstDayDeviceNum).firstDayPayAmount(firstDayPayAmount)
                                            .firstDayOrderNum(firstDayOrderNum).ts(tradeOrder.getTs()).createTime(createTime).winStartTime(startTime)
                                            .winEndTime(endTime).labCode(orderFiledBean.getLabCode()).labGroupCode(orderFiledBean.getLabGroupCode()).build();

                                    if (!firstPay) {
                                        orderFiledBean.setExistPay(true);
                                        lastPayState.update(orderFiledBean);
                                    }

                                } else if (tradeOrder.getState() == 5) {
                                    //退款数据
                                    orderBean = TradeOrderBean.builder().os(orderFiledBean.getOs()).code(orderFiledBean.getCode()).product_id(tradeOrder.getProductId())
                                            .wayCode(tradeOrder.getWayCode()).payAmount(BigDecimal.ZERO).refundAmount(tradeOrder.getRefundAmount())
                                            .taxAmount(BigDecimal.ZERO).totalAmount(BigDecimal.ZERO).orderNum(0L).deviceNum(0L).ts(tradeOrder.getTs())
                                            .firstDayDeviceNum(0L).firstDayPayAmount(BigDecimal.ZERO).firstDayOrderNum(0L).createTime(createTime).winStartTime(startTime)
                                            .winEndTime(endTime).labCode(orderFiledBean.getLabCode()).labGroupCode(orderFiledBean.getLabGroupCode()).build();
                                }
                                //补全数据
                                if (StringUtils.isEmpty(orderBean.getCode())) {
                                    orderBean.setCode("note");
                                }

                                if (StringUtils.isEmpty(orderBean.getOs())) {
                                    if ("APPLE_APP".equals(orderBean.getWayCode())) {
                                        orderBean.setOs("IOS");
                                    } else {
                                        orderBean.setOs("Android");
                                    }
                                }

                                out.collect(orderBean);
                            } else {
                                if (tradeOrder.getStartWay() == 1) {
                                    if (orderFiledBean == null) {
                                        orderFiledBean = new OrderFiledBean();
                                    }
                                    //第一次启动重新更新
                                    Timestamp registerTime = tradeOrder.getRegisterTime();
                                    String curDt = DateFormatUtil.toDate(registerTime.getTime());
                                    orderFiledBean.setCode(tradeOrder.getCode());
                                    orderFiledBean.setOs(tradeOrder.getOs());
                                    orderFiledBean.setLabCode(tradeOrder.getLabCode());
                                    orderFiledBean.setLabGroupCode(tradeOrder.getLabGroupCode());
                                    orderFiledBean.setDate(curDt);
                                    orderFiledBean.setDeviceId(tradeOrder.getDeviceId());
                                    orderFiledBean.setExistPay(false);

                                    lastPayState.update(orderFiledBean);
                                } else {
                                    if (orderFiledBean == null) {
                                        Timestamp yesToday = new Timestamp(DateUtils.addDays(new Date(), -1).getTime());
                                        String yesTodayStr = DateFormatUtil.toDate(yesToday.getTime());
                                        orderFiledBean = OrderFiledBean.builder().os(tradeOrder.getOs()).code(tradeOrder.getCode())
                                                .deviceId(tradeOrder.getDeviceId()).labCode(tradeOrder.getLabCode())
                                                .labGroupCode(tradeOrder.getLabGroupCode()).date(yesTodayStr).existPay(false).build();
                                        lastPayState.update(orderFiledBean);
                                    } else if (orderFiledBean != null && StringUtils.isEmpty(orderFiledBean.getLabCode())) {
                                        orderFiledBean.setLabCode(tradeOrder.getLabCode());
                                        orderFiledBean.setLabGroupCode(tradeOrder.getLabGroupCode());
                                        lastPayState.update(orderFiledBean);
                                    }
                                }
                            }

                        }
                    });

            // 8.将数据写出到ClickHouse
            userPayOrderDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_trade_order_window values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

            // 9.启动任务
            env.execute("DwsTradeOrderWindow");
        } catch (Exception e) {
            logger.error("DwsTradeOrderWindow error", e);
        }
    }

    /**
     * 订单状态：（成功、失败）
     */
    private static final List<Integer> AVAILABLE_STATE = Arrays.asList(2, 5);

    private static boolean validateDb(DwdTradeOrder v) {
//        if (!v.getType().equals("update")) {
//            return false;
//        }

//        if (v.getOldState() == null) {
//            return false;
//        }

        if (!v.getDatabase().equals("note_data")) {
            return false;
        }

        if (!v.getTable().equals("t_pay_order")) {
            return false;
        }

        return AVAILABLE_STATE.contains(v.getState());
    }

}
